package server

import (
	"context"
)

// 消息管理组件，生命周期和 server 实例保持一致。
type MessageManager struct {
	sessionManager *sessionManager
	sessionBuilder *SessionBuilder
	writeQueue     chan *MessageJob
	readQueue      chan *MessageJob
	ctx            context.Context
}

func NewMessageManager(sessionManager *sessionManager, sessionBuilder *SessionBuilder, ctx context.Context) *MessageManager {
	return &MessageManager{
		sessionManager: sessionManager,
		ctx:            ctx,
		sessionBuilder: sessionBuilder,
	}
}

func (x *MessageManager) init() {
	x.readQueue = make(chan *MessageJob, x.sessionBuilder.ReadQueueSize)
	x.writeQueue = make(chan *MessageJob, x.sessionBuilder.WriteQueueSize)
	go x.read()
	go x.write()
}

func (x *MessageManager) GroupSend(data []byte, filter func(session ISession) bool) {
	x.sessionManager.IterateSession(func(session ISession) {
		if filter == nil || filter(session) {
			x.Write(session, data)
		}
	})
}

func (x *MessageManager) Broadcast(data []byte) {
	x.GroupSend(data, nil)
}

func (x *MessageManager) Write(session ISession, data []byte) {
	x.writeQueue <- NewMessageJob(session, data)
}

func (x *MessageManager) Read(session ISession, data []byte) {
	x.readQueue <- NewMessageJob(session, data)
}

func (x *MessageManager) read() {
	for job := range x.readQueue {
		job.session.read(job.data)
	}
}

func (x *MessageManager) write() {
	for job := range x.writeQueue {
		job.session.write(job.data)
	}
}

func (x *MessageManager) Close() {
	close(x.writeQueue)
	close(x.readQueue)
}
